package com.wxzhou.wangzhuan.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.CollectionType;
import com.wxzhou.wangzhuan.constants.CacheNames;
import com.wxzhou.wangzhuan.entity.AdLog;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.pubsub.ReactivePubSubCommands;
import io.quarkus.runtime.StartupEvent;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Observes;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.util.List;

/**
 * 广告上报消费者
 * 废弃,不走订阅模式!
 */
@ApplicationScoped
public class AdLogConsumerService {

    private static final Logger LOG = Logger.getLogger(AdLogConsumerService.class);

    private ReactivePubSubCommands<String> pubSubCommands;

    private CollectionType collectionType;

    @Inject
    ReactiveRedisDataSource reactiveRedisDataSource;

    @Inject
    ObjectMapper objectMapper;

    @Inject
    AdService adService;


    void onStart(@Observes StartupEvent ev) {
        LOG.info("开始初始化Redis订阅服务...");
        this.pubSubCommands = reactiveRedisDataSource.pubsub(String.class);
        this.collectionType = objectMapper.getTypeFactory().constructCollectionType(List.class, AdLog.class);
        if (this.pubSubCommands == null) {
            LOG.error("获取Redis PubSub命令对象失败！");
            return;
        }

        LOG.info("准备订阅频道: " + CacheNames.BATCH_ADLOG_CHANNEL);

        // 订阅批量广告数据完成状态频道
        try {
            pubSubCommands.subscribe(CacheNames.BATCH_ADLOG_CHANNEL).subscribe().with(
                    // 处理接收到的消息
                    message -> {
                        LOG.info("接收到批量广告数据状态通知: " + message);
                        System.out.println("批量广告数据状态通知: " + message);
                        //查询这一批广告数据清单
                        List<AdLog> adLogs = convertJsonToAdLog(message);
                        if (adLogs != null && !adLogs.isEmpty()) {
                            LOG.infof("总条数: %d", adLogs.size());
                            adService.batchInsert(adLogs).subscribe().with(integer -> {
                                LOG.infof("插入成功，影响行数：" + integer);
                            }, throwable -> {
                                LOG.errorf(throwable, "批量执行插入广告数据发生异常");
                            });

                        } else {
                            LOG.warn("未找到广告数据清单数据");
                        }
                    },
                    // 处理订阅过程中的错误
                    error -> {
                        LOG.error("订阅频道失败: " + CacheNames.BATCH_ADLOG_CHANNEL, error);
                        System.err.println("订阅频道失败: " + CacheNames.BATCH_ADLOG_CHANNEL + ", 错误: " + error.getMessage());
                    },
                    // 订阅完成的回调
                    () -> {
                        LOG.info("订阅已完成: " + CacheNames.BATCH_ADLOG_CHANNEL);
                        System.out.println("订阅已完成: " + CacheNames.BATCH_ADLOG_CHANNEL);
                    });

            LOG.info("初始化Redis订阅服务成功，已订阅频道: " + CacheNames.BATCH_ADLOG_CHANNEL);
        } catch (Exception e) {
            LOG.error("订阅Redis频道时发生异常", e);
            System.err.println("订阅Redis频道时发生异常: " + e.getMessage());
        }
    }

    /**
     * 将json字符串转换成集合
     *
     * @param json
     * @return
     */
    public List<AdLog> convertJsonToAdLog(String json) {
        if (json == null || json.isEmpty()) {
            LOG.warn("json内容为空");
            return null;
        }
        try {
            List<AdLog> result = objectMapper.readValue(json, collectionType);
            LOG.info("JSON解析成功，共解析到 " + (result != null ? result.size() : 0) + " 条记录");
            return result;
        } catch (Exception e) {
            LOG.errorf(e, "解析广告数据记录json字符串失败！ %s", json);
            return null;
        }

    }

}
